//
// Created by EDY on 2025/10/11.
//

#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include "connectionManger.h"
#include "../memoryPool/e_palloc.h"

extern ConnectionManger m_conn_manager;
static int next_conn_id = 1;
extern struct mp_pool_s *m_mp;

int connection_manager_init(void) {
    memset(&m_conn_manager,0,sizeof(m_conn_manager));

    m_conn_manager.connections_by_fd = NULL;
    m_conn_manager.connections_by_id = NULL;

    //初始化读写锁
    if (pthread_rwlock_init(&m_conn_manager.hash_lock,NULL) != 0) {
        return -1;
    }

    atomic_store(&m_conn_manager.next_conn_id,1);

    atomic_store(&m_conn_manager.global_stats.total_connections,0);
    atomic_store(&m_conn_manager.global_stats.total_bytes_transferred,0);
    atomic_store(&m_conn_manager.global_stats.max_concurrent,0);
    atomic_store(&m_conn_manager.global_stats.active_connections,0);
    m_conn_manager.global_stats.start_time = time(NULL);

    printf("Connection Manager with uthash initialized successfully\n");
    return 0;
}
void connection_manager_destroy(void) {
    pthread_rwlock_destroy(&m_conn_manager.hash_lock);

    //清理所有链接
    Connection *conn,*tmp;
    HASH_ITER(hh,m_conn_manager.connections_by_fd,conn,tmp) {
        HASH_DEL(m_conn_manager.connections_by_fd,conn);

        //清理设备上下文
        if (conn->devcie_ctx) {
            if (conn->devcie_ctx->sock_fd > 0 ) {
                close(conn->devcie_ctx->sock_fd);
            }
            mp_free(m_mp,conn->devcie_ctx);
        }

        pthread_mutex_destroy(&conn->conn_lock);
        mp_free(m_mp,conn);
    }

    //清理链接ID映射表
    ConnectionIdMap *id_map,*id_tmp;
    HASH_ITER(hh,m_conn_manager.connections_by_id,id_map,id_tmp) {
        HASH_DEL(m_conn_manager.connections_by_id,id_map);
        mp_free(m_mp,id_map);
    }

    pthread_rwlock_unlock(&m_conn_manager.hash_lock);
    pthread_rwlock_destroy(&m_conn_manager.hash_lock);

    printf("Connection manager destroyed\n");

}

Connection* connection_acquire(int sock_fd) {
    if (sock_fd < 0 ) {
        return NULL;
    }

    pthread_rwlock_wrlock(&m_conn_manager.hash_lock);

    //检查连接是否已存在
    Connection *existing_conn;
    HASH_FIND_INT(m_conn_manager.connections_by_fd,&sock_fd,existing_conn);
    if (existing_conn) {
        pthread_rwlock_unlock(&m_conn_manager.hash_lock);
        printf("Connection already exists\n");
        return NULL;
    }

    //检查连接数限制
    unsigned int active_count =
        atomic_load(&m_conn_manager.global_stats.active_connections);
    if (active_count >= MAX_CONNECTIONS) {
        pthread_rwlock_unlock(&m_conn_manager.hash_lock);
        printf("Connection pool exhausted (MAX : %d)\n",MAX_CONNECTIONS);
        return NULL;
    }

    //创建新连接
    Connection *conn = (Connection *)mp_malloc(m_mp,sizeof(Connection));
    if (!conn) {
        pthread_rwlock_unlock(&m_conn_manager.hash_lock);
        printf("malloc faslse,failed to create new connection \n");
        return NULL;
    }

    //初始化连接
    memset(conn,0,sizeof(Connection));
    conn->sock_fd = sock_fd;
    conn->conn_id = atomic_fetch_add(&m_conn_manager.next_conn_id,1);

    //创建设备上下文
    conn->devcie_ctx = (DeviceContext *)mp_malloc(m_mp,sizeof(DeviceContext));
    if (!conn->devcie_ctx) {
        mp_free(m_mp,conn);
        pthread_rwlock_unlock(&m_conn_manager.hash_lock);
        return NULL;
    }
    memset(conn->devcie_ctx,0,sizeof(DeviceContext));
    conn->devcie_ctx->sock_fd = sock_fd;
    conn->devcie_ctx->state = DEVICE_STATE_UNREGISTERED;

    conn->state = CONN_STATE_CONNECTED;
    conn->connect_time = time(NULL);
    conn->last_active = time(NULL);

    //初始化统计信息
    memset(&conn->stats,0,sizeof(ConnectionStats));
    conn->stats.first_connect_times = time(NULL);

    //初始化连接锁
    if (pthread_mutex_init(&conn->conn_lock,NULL) != 0) {
        mp_free(m_mp,conn->devcie_ctx);
        mp_free(m_mp,conn);
        pthread_rwlock_unlock(&m_conn_manager.hash_lock);
        return NULL;
    }

    //添加到按fd索引的哈希表
    HASH_ADD_INT(m_conn_manager.connections_by_fd,sock_fd,conn);

    //创建并添加到按ID索引的哈希表
    ConnectionIdMap *id_map = (ConnectionIdMap *)mp_malloc(m_mp,sizeof(ConnectionIdMap));
    if (!id_map) {
        HASH_DEL(m_conn_manager.connections_by_fd,conn);
        mp_free(m_mp,conn->devcie_ctx);
        mp_free(m_mp,conn);
        pthread_rwlock_unlock(&m_conn_manager.hash_lock);
        return NULL;
    }
    if (id_map) {
        id_map->conn_id = conn->conn_id;
        id_map->connection = conn;
        HASH_ADD_INT(m_conn_manager.connections_by_id,conn_id,id_map);
    }

    pthread_rwlock_unlock(&m_conn_manager.hash_lock);

    atomic_fetch_add(&m_conn_manager.global_stats.total_connections,1);
    uint32_t new_active =
        atomic_fetch_add(&m_conn_manager.global_stats.active_connections,1) + 1;

    //更新最大并发数
    uint32_t current_max = atomic_load(&m_conn_manager.global_stats.max_concurrent);
    while (new_active > current_max) {
        if (atomic_compare_exchange_weak(&m_conn_manager.global_stats.max_concurrent,
                                    &current_max,new_active)){
            break;
        }

    }
    printf("Connection acquired : ID = %d ,FD = %d (Acticve : %u)\n",
        conn->conn_id,sock_fd,new_active);

    return conn;


}
int connection_release(Connection *conn) {
    if (!conn) {
        return -1;
    }

    pthread_rwlock_wrlock(&m_conn_manager.hash_lock);

    //从按FD索引的哈希表中删除
    Connection *found_conn;
    HASH_FIND_INT(m_conn_manager.connections_by_fd,&conn->sock_fd,found_conn);
    if (found_conn == conn) {
        HASH_DEL(m_conn_manager.connections_by_fd,conn);
    }

    //按从id索引的哈希表中删除
    ConnectionIdMap *id_map;
    HASH_FIND_INT(m_conn_manager.connections_by_id,&conn->conn_id,id_map);
    if (id_map) {
        HASH_DEL(m_conn_manager.connections_by_id,id_map);
        mp_free(m_mp,id_map);
    }

    pthread_rwlock_unlock(&m_conn_manager.hash_lock);

    printf("Releasing connection : FD = %d,Id = %d\n",conn->sock_fd,conn->conn_id);

    //清理资源
    if (conn->devcie_ctx) {
        if (conn->devcie_ctx->sock_fd > 0 ) {
            close(conn->devcie_ctx->sock_fd);
        }
        mp_free(m_mp,conn->devcie_ctx);
    }

    pthread_mutex_destroy(&conn->conn_lock);
    mp_free(m_mp,conn);

    atomic_fetch_sub(&m_conn_manager.global_stats.active_connections,1);

    return 0;

}
// int connection_bind_device(Connection *conn,Device *device);
// int conection_unbind_device(Connection *conn);
Connection *connection_find_by_fd(int sock_fd) {
    if (sock_fd < 0) {
        return NULL;
    }

    pthread_rwlock_rdlock(&m_conn_manager.hash_lock);

    Connection *conn;
    HASH_FIND_INT(m_conn_manager.connections_by_fd,&sock_fd,conn);

    pthread_rwlock_unlock(&m_conn_manager.hash_lock);
    return conn;
}
Connection *connection_find_by_id(int conn_id) {
    if (conn_id <= 0) {
        return NULL;
    }

    pthread_rwlock_rdlock(&m_conn_manager.hash_lock);

    ConnectionIdMap *id_map;
    HASH_FIND_INT(m_conn_manager.connections_by_id,&conn_id,id_map);

    Connection *conn = id_map ? id_map->connection : NULL;

    pthread_rwlock_unlock(&m_conn_manager.hash_lock);

    return conn;

}
int connection_update_stats(Connection *conn,size_t bytes_in , size_t bytes_out) {
    if (!conn) return -1;

    pthread_mutex_lock(&conn->conn_lock);

    conn->stats.bytes_received += bytes_in;
    conn->stats.bytes_sent += bytes_out;
    if (bytes_in > 0 ) conn->stats.packets_received++;
    if (bytes_out > 0) conn->stats.packets_sent++;

    conn->last_active = time(NULL);
    pthread_mutex_unlock(&conn->conn_lock);
    atomic_fetch_add(&m_conn_manager.global_stats.total_bytes_transferred,bytes_in+bytes_out);
    return 0;
}
int connection_check_timeout(void) {
    time_t now = time(NULL);
    int timeout_count = 0;

    pthread_rwlock_rdlock(&m_conn_manager.hash_lock);

    Connection *conn,*tmp;
    HASH_ITER(hh,m_conn_manager.connections_by_fd,conn,tmp) {
        pthread_mutex_lock(&conn->conn_lock);

        if (now - conn->last_active > CONN_TIMEOUT) {
            printf("Connection timeout ID = %d,fd = %d\n",conn->conn_id,conn->sock_fd);
            conn->state = CONN_STATE_TIMEOUT;
            timeout_count++;
        }
        pthread_mutex_unlock(&conn->conn_lock);
        pthread_rwlock_unlock(&m_conn_manager.hash_lock);
        return timeout_count;
    }
    pthread_rwlock_unlock(&m_conn_manager.hash_lock);
}

int connection_foreach(connection_callback_t callback, void *user_data) {
    if (!callback) return -1;
    pthread_rwlock_rdlock(&m_conn_manager.hash_lock);

    Connection *conn,*tmp;
    int processed = 0;
    HASH_ITER(hh,m_conn_manager.connections_by_fd,conn,tmp) {
        if (callback(conn,user_data) == 0) {
            processed++;
        }
    }

    pthread_rwlock_unlock(&m_conn_manager.hash_lock);

    return processed;


}
// int connection_send_heartbeat(Connection *conn);
void Connection_print_stats(void) {
    uint32_t total = atomic_load(&m_conn_manager.global_stats.total_connections);
    uint32_t active = atomic_load(&m_conn_manager.global_stats.active_connections);
    uint32_t max_concurrent = atomic_load(&m_conn_manager.global_stats.max_concurrent);
    uint64_t total_bytes = atomic_load(&m_conn_manager.global_stats.total_bytes_transferred);

    pthread_rwlock_rdlock(&m_conn_manager.hash_lock);

    // 获取哈希表当前大小
    unsigned int hash_count = HASH_COUNT(m_conn_manager.connections_by_fd);

    pthread_rwlock_unlock(&m_conn_manager.hash_lock);

    printf("\n=== Connection Manager Statistics (uthash) ===\n");
    printf("Total Connections: %u\n", total);
    printf("Active Connections: %u\n", active);
    printf("Hash Table Size: %u\n", hash_count);
    printf("Max Concurrent: %u\n", max_concurrent);
    printf("Total Bytes Transferred: %lu\n", total_bytes);
    printf("Uptime: %ld seconds\n", time(NULL) - m_conn_manager.global_stats.start_time);
    printf("============================================\n\n");
}

ConnectionState connection_get_state(Connection *conn) {
    if (!conn) return CONN_STATE_ERROR;

    pthread_mutex_lock(&conn->conn_lock);
    ConnectionState state = conn->state;
    pthread_mutex_unlock(&conn->conn_lock);

    return state;
}
